package cn.com.cennavi.services;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.io.Serializable;
import java.net.MalformedURLException;
import java.nio.charset.Charset;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;
import org.geotools.data.FeatureSource;
import org.geotools.data.shapefile.ShapefileDataStore;
import org.geotools.feature.FeatureCollection;
import org.geotools.feature.FeatureIterator;
import org.opengis.feature.simple.SimpleFeature;
import org.opengis.feature.simple.SimpleFeatureType;
import org.opengis.feature.type.AttributeDescriptor;

import scala.Tuple2;
import cn.com.cennavi.bean.MineLabODBean;
import cn.com.cennavi.bean.OD_HotSpot_OutDataBean;
import cn.com.cennavi.bean.ShpFileBean;
import cn.com.cennavi.bean.TrajectoryBean;
import cn.com.cennavi.config.GlobalConfig;
import cn.com.cennavi.config.MineLabODConfig;
import cn.com.cennavi.factory.MineLabServiceFactory;
import cn.com.cennavi.interfaces.IMineLabModel;
import cn.com.cennavi.util.MapUtil;
import cn.com.cennavi.util.ModelInPutUtil;

import com.vividsolutions.jts.geom.Envelope;
import com.vividsolutions.jts.geom.GeometryFactory;
import com.vividsolutions.jts.geom.MultiPolygon;
import com.vividsolutions.jts.geom.Polygon;
import com.vividsolutions.jts.geom.PrecisionModel;
import com.vividsolutions.jts.io.WKTReader;

/**
 * 
 * 
 * @ClassName: ODAnalysis
 * @Description: TODO(minelab中的od模块)
 * @author buf
 * @date 2017年2月7日 上午10:47:53
 *
 */
public class ODAnalysis extends IMineLabModel implements Serializable {
	/**
	 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么)
	 */
	private static final long serialVersionUID = 1L;
	private static MineLabODConfig mineLabODConfig = null;
	private static SparkConf sparkConf = null;
	private String userId;
	private String modelId;
	private String modelminId;
	private String dataSourse;
	private String dataClass;
	private String startDateTime;
	private String endDateTime;
	private int spatialFlag;
	private int meshCode;
	private int cityCode;
	private int provinceCode;
	private int timegranularityFlag;
	private float timegranularity;
	private int spatialgranularityFlag;
	private String geoshapePath;
	private String tsakID;
	private int outflag;
	private String outPath;
	private int countyCode;

	public ODAnalysis(String userId, String modelId, String modelminId, String dataSourse, String dataClass, String startDateTime, String endDateTime, int spatialFlag, int meshCode, int countyCode, int cityCode, int provinceCode, int timegranularityFlag, float timegranularity, int spatialgranularityFlag, String geoshapePath, int outflag, String outPath, String tsakID) {
		// TODO Auto-generated constructor stub
		this.userId = userId;
		this.modelId = modelId;
		this.modelminId = modelminId;
		this.dataSourse = dataSourse;
		this.dataClass = dataClass;
		this.startDateTime = startDateTime;
		this.endDateTime = endDateTime;
		this.spatialFlag = spatialFlag;
		this.meshCode = meshCode;
		this.cityCode = cityCode;
		this.provinceCode = provinceCode;
		this.timegranularityFlag = timegranularityFlag;
		this.timegranularity = timegranularity;
		this.spatialgranularityFlag = spatialgranularityFlag;
		this.geoshapePath = geoshapePath;
		this.tsakID = tsakID;
		this.outflag = outflag;
		this.outPath = outPath;
		this.countyCode = countyCode;
	}

	public void doJob() {

		System.out.println("start to do the od job!");
		 System.setProperty("hadoop.home.dir", "d:\\hadoop-2.4.0");

		// spark集群配置项,算法运行环境默认配置
		sparkConf = new GlobalConfig().sparkconf;
		mineLabODConfig = new MineLabODConfig();

		// 算法运行传入参数配置,用构造函数中的参数替换默认配置文件中的参数，没替代的还是用默认配置
		reflushConfigArgs();
		JavaSparkContext sc = new JavaSparkContext(sparkConf);

		// 根据时间范围和数据源类型获取所有输入文件的路径,这里默认数据是保存在hdfs上的
		try {

			// 读取地图shp文件
			// 读取二次网格与county对照表
			Map<Integer, Set<Integer>> meshCountyMap = getMeshCountyMap(mineLabODConfig.meshToCountyCorrespondFile);
			Broadcast<Map<Integer, Set<Integer>>> meshCountyMapBroad = sc.broadcast(meshCountyMap);

			// 获取最小粒度地图shp,生成odNum结果时的空间最小粒度
			List<ShpFileBean> spatialgranularityMap = ("".equals(mineLabODConfig.spatialgranularityGeoshapePath) || mineLabODConfig.spatialgranularityGeoshapePath == null) ? null : getMMParameter(mineLabODConfig.spatialgranularityGeoshapePath);
			Broadcast<List<ShpFileBean>> spatialgranularityBroad = (spatialgranularityMap == null ? null : sc.broadcast(spatialgranularityMap));

			HashMap<String, String> filePathsList = null;
			try {
				filePathsList = ModelInPutUtil.getFilePaths(mineLabODConfig, meshCountyMap);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			System.out.println("filePaths--------------------" );
			for (Map.Entry<String, String> e : filePathsList.entrySet()) {
				System.out.println("filePaths:" + e.getKey());
			}
			System.out.println("filePaths--------------------" );
			// if (filePathsList == null) {
			// System.out.println("输入文件为空！");
			// System.exit(0);
			// }
			// JavaRDD<MineLabODBean> unionRdd = null;
			// int num = 0;
			// for (Map.Entry<String, String> e : filePathsList.entrySet()) {
			// String filePaths = e.getKey();
			// String[] dataSourse_dataClass = e.getValue().split("\\_");
			// System.out.println("niuxinzan:filePaths:" + filePaths +
			// ":dataSourse_dataClass:" + e.getValue());
			//
			// String dataSourse = dataSourse_dataClass[0];
			// System.out.println("niuxinzan:dataSourse:" + dataSourse);
			// // 读取数据
			// JavaRDD<String> rdd = sc.textFile(filePaths);
			// // 解析数据,将数据解析成OD模块需要的数据特征对象
			// JavaRDD<MineLabODBean> objectData = new
			// DataParseService(dataSourse).doJob(rdd).filter(new
			// Function<MineLabODBean, Boolean>() {
			// @Override
			// public Boolean call(MineLabODBean arg0) throws Exception {
			// // TODO Auto-generated method stub
			// return arg0 == null ? false : true;
			// }
			// });
			// if (num == 0) {
			// unionRdd = objectData;
			// num++;
			// } else {
			// unionRdd = unionRdd.union(objectData);
			//
			// }
			// }
			// // // 时空筛选模块，特征筛选，将数据解析成od模块子模块需要的特征对象
			// JavaRDD<MineLabODBean> mineLabODBean = unionRdd.map(new
			// DataScreen(meshCountyMapBroad, mineLabODConfig)).filter(new
			// Function<MineLabODBean, Boolean>() {
			//
			// @Override
			// public Boolean call(MineLabODBean arg0) throws Exception {
			// // TODO Auto-generated method stub
			// return arg0 == null ? false : true;
			// }
			// });
			// // 调用计算模块进行计算
			// getODModel(mineLabODBean, spatialgranularityBroad,
			// mineLabODConfig, meshCountyMapBroad);

		} catch (com.vividsolutions.jts.io.ParseException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} finally {
			sc.stop();
			System.clearProperty("spark.driver.port");
		}

	}

	private Map<Integer, Set<Integer>> getMeshCountyMap(String spatialgranularityGeoshapePath) throws IOException {
		// TODO Auto-generated method stub
		Map<Integer, Set<Integer>> returnMap = new HashMap<Integer, Set<Integer>>();
		BufferedReader br = new BufferedReader(new FileReader(new File(spatialgranularityGeoshapePath)));
		String line = "";
		while ((line = br.readLine()) != null) {
			String[] arr = line.split(",");
			// System.out.println(arr.length);
			Integer countyId = Integer.parseInt(arr[0]);
			for (int i = 2; i < arr.length; i++) {
				// System.out.println(arr[i]);
				Integer mesh = Integer.parseInt(arr[i]);
				if (returnMap.containsKey(mesh)) {
					returnMap.get(mesh).add(countyId);
				} else {
					Set<Integer> dataSet = new HashSet<Integer>();
					dataSet.add(countyId);
					returnMap.put(mesh, dataSet);
				}
			}
			// System.out.println("-------------------");
		}
		return returnMap;
	}

	/**
	 * 获取一个整体的区域MultiPolygon
	 * 
	 * @Description
	 * @author Thinkpad
	 * @param spatialgranularityGeoshapePath
	 * @return
	 */
	private List<ShpFileBean> getMMParameter1(String spatialgranularityGeoshapePath) {
		// TODO Auto-generated method stub
		return null;
	}

	/**
	 * 
	 * @Description
	 * @author buf
	 */
	private void reflushConfigArgs() {
		// TODO Auto-generated method stub
		// --------------------------------spark任务名称------------------------------------//
		sparkConf.setAppName(tsakID);
		// --------------------------------用户信息------------------------------------//
		mineLabODConfig.userId = userId;
		// --------------------------------模型参数------------------------------------//
		mineLabODConfig.modelId = modelId;
		mineLabODConfig.modelminId = modelminId;
		// --------------------------------数据选取------------------------------------//
		// 数据源
		mineLabODConfig.dataSourse = dataSourse;
		// 数据类型
		mineLabODConfig.dataClass = dataClass;
		// --------------------------------时间范围------------------------------------//
		// 开始日期
		mineLabODConfig.startDate = startDateTime;
		// 结束日期
		mineLabODConfig.endDate = endDateTime;

		// --------------------------------空间范围------------------------------------//
		mineLabODConfig.spatialFlag = spatialFlag;
		// 城市代码
		mineLabODConfig.meshCode = meshCode;
		mineLabODConfig.countyCode = countyCode;
		mineLabODConfig.cityCode = cityCode;
		mineLabODConfig.province = provinceCode;
		// mineLabODConfig.spatialGeoShpFilePath = spatialGeoShpFilePath;

		// ---------------------------------时间粒度------------------------------------//
		// 时间粒度单位：0分钟，1 小时，2天，3月，4年
		if (timegranularityFlag != -1)
			mineLabODConfig.timegranularityFlag = timegranularityFlag;
		// 时间粒度
		if (timegranularity != -1)
			mineLabODConfig.timegranularity = timegranularity;

		// ---------------------------------空间粒度-------------------------------------//
		// 空间粒度标志：
		// 1代表最小粒度为二次网格
		// 2代表最小粒度为不规则形状，目前不规则形状为地图边界图
		if (spatialgranularityFlag != -1)
			mineLabODConfig.spatialgranularityFlag = spatialgranularityFlag;
		// 最小粒度为不规则形状3时，地图属性文件路径
		// 区域形状
		if (!geoshapePath.equals("-1"))
			mineLabODConfig.spatialgranularityGeoshapePath = geoshapePath;

		// ----------------------------------输出形式------------------------------------//
		mineLabODConfig.outFlag = outflag;
		mineLabODConfig.outpath = outPath;

		System.out.println("----------config------------");
		System.out.println(mineLabODConfig.userId);
		// --------------------------------模型参数------------------------------------//
		System.out.println(mineLabODConfig.modelId);
		System.out.println(mineLabODConfig.modelminId);
		// --------------------------------数据选取------------------------------------//
		// 数据源
		System.out.println(mineLabODConfig.dataSourse);
		// 数据类型
		System.out.println(mineLabODConfig.dataClass);
		// --------------------------------时间范围------------------------------------//
		// 开始日期
		System.out.println(mineLabODConfig.startDate);
		// 结束日期
		System.out.println(mineLabODConfig.endDate);

		// --------------------------------空间范围------------------------------------//
		System.out.println(mineLabODConfig.spatialFlag);
		// 城市代码
		System.out.println(mineLabODConfig.countyCode);
		System.out.println(mineLabODConfig.cityCode);
		System.out.println(mineLabODConfig.province);
		// System.out.println(mineLabODConfig.spatialGeoShpFilePath);

		// ---------------------------------时间粒度------------------------------------//
		// 时间粒度,单位小时
		System.out.println(mineLabODConfig.timegranularityFlag);
		System.out.println(mineLabODConfig.timegranularity);

		// ---------------------------------空间粒度-------------------------------------//
		// 空间粒度标志：
		// 0代表最小粒度为二次网格
		// 1代表最小粒度为不规则形状，目前不规则形状为地图边界图
		System.out.println(mineLabODConfig.spatialgranularityFlag);
		// 最小粒度为不规则形状1时，地图属性文件路径
		// 区域形状
		System.out.println(mineLabODConfig.spatialgranularityGeoshapePath);

		// ----------------------------------输出形式------------------------------------//
		System.out.println(mineLabODConfig.outFlag);
		System.out.println(mineLabODConfig.outpath);
		System.out.println("----------------------");

	}

	/**
	 * 读取文件
	 */
	public static JavaRDD<MineLabODBean> doJob(JavaRDD<String> filesData) {
		if (mineLabODConfig.dataSourse.equals("trajectory")) {
			JavaRDD<MineLabODBean> rdd = filesData.map(new Function<String, MineLabODBean>() {

				@Override
				public MineLabODBean call(String arg0) throws Exception {
					// TODO Auto-generated method stub
					try {
						TrajectoryBean tb = (TrajectoryBean) TrajectoryBean.parseData(arg0);
						if (tb == null) {
							return null;
						} else {
							MineLabODBean mb = new MineLabODBean();
							mb.oTime = tb.getOTime();
							mb.dTime = tb.getDTime();
							mb.oLongitude = tb.getOLongitude();
							mb.oLatitude = tb.getOLatitude();
							mb.dLongitude = tb.getDLongitude();
							mb.dLatitude = tb.getDLatitude();
							return mb;
						}

					} catch (Exception e) {
						// TODO: handle exception
						e.printStackTrace();
					}
					return null;
				}
			});
			return rdd;
		} else {
			// 扩展项
		}
		return null;

	}

	private List<ShpFileBean> getMMParameter(String inputPath_MP) throws com.vividsolutions.jts.io.ParseException {
		// TODO Auto-generated method stub
		List<ShpFileBean> result1 = new ArrayList<ShpFileBean>();
		// 读取地图文件
		ShapefileDataStore shpDataStore = null;
		try {
			shpDataStore = new ShapefileDataStore(new File(inputPath_MP).toURI().toURL());
			shpDataStore.setStringCharset(Charset.forName("gbk"));
			// 文件名称
			String typeName = shpDataStore.getTypeNames()[0];
			FeatureSource<SimpleFeatureType, SimpleFeature> featureSource = null;
			featureSource = (FeatureSource<SimpleFeatureType, SimpleFeature>) shpDataStore.getFeatureSource(typeName);
			FeatureCollection<SimpleFeatureType, SimpleFeature> result = featureSource.getFeatures();
			SimpleFeatureType schema = result.getSchema(); // schema
			List<AttributeDescriptor> columns = schema.getAttributeDescriptors();
			FeatureIterator<SimpleFeature> itertor = result.features();

			ShpFileBean sfb = null;
			while (itertor.hasNext()) {
				SimpleFeature feature = itertor.next();
				for (AttributeDescriptor attributeDes : columns) {
					String attributeName = attributeDes.getName().toString();// attribute
					// System.out.println(attributeName);
					if (attributeName.equals("the_geom")) {
						sfb = new ShpFileBean();
						int srid = 4326;
						GeometryFactory geometryFactory = new GeometryFactory(new PrecisionModel(), srid);
						WKTReader reader = new WKTReader(geometryFactory);
						MultiPolygon mpolygon = (MultiPolygon) reader.read(feature.getAttribute(attributeName).toString());
						sfb.shpPoint = mpolygon;
					}
					if (attributeName.equals("NO")) {
						sfb.shpId = (Integer) feature.getAttribute(attributeName);
						result1.add(sfb);
					}
				}
			}
			itertor.close();
		} catch (MalformedURLException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return result1;
	}

	/**
	 * 获取对象模型
	 * 
	 * @param meshCountyMapBroad
	 */
	private void getODModel(JavaRDD<MineLabODBean> mineLabODNumBean, Broadcast<List<ShpFileBean>> mapBroad, final MineLabODConfig mineLabODConfig, Broadcast<Map<Integer, Set<Integer>>> meshCountyMapBroad) {
		// TODO Auto-generated method stub
		switch (mineLabODConfig.modelminId) {
		case "odNum":
			// 如果是统计指定时空范围内od数，则执行ODAnalysis_Trajectory_ODNum服务，包括数据解析，实现统计等等
			// 0代表最小粒度为二次网格
			// 1自定义粒度，目前不规则形状为地图边界图
			JavaPairRDD<String, Integer> rdd1 = mineLabODNumBean.flatMapToPair(new ODAnalysisTrajectoryODNum(mapBroad, mineLabODConfig.spatialFlag, mineLabODConfig.meshCode, mineLabODConfig.countyCode, mineLabODConfig.cityCode, mineLabODConfig.province, mineLabODConfig.timegranularityFlag, mineLabODConfig.timegranularity, mineLabODConfig.dataSourse, meshCountyMapBroad, mineLabODConfig.spatialgranularityFlag));
			JavaPairRDD<String, Integer> rdd2 = rdd1.reduceByKey(new Function2<Integer, Integer, Integer>() {

				public Integer call(Integer arg0, Integer arg1) throws Exception {
					// TODO Auto-generated method stub
					return arg0 + arg1;
				}
			});
			JavaRDD<String> rdd3 = rdd2.map(new Function<Tuple2<String, Integer>, String>() {

				public String call(Tuple2<String, Integer> arg0) throws Exception {
					// TODO Auto-generated method stub
					if (!arg0._1.contains("errorData")) {
						String data = arg0._1;
						int odNum = arg0._2;
						String[] dataStr = data.split(",");
						dataStr[19] = odNum + "";
						StringBuffer sb = new StringBuffer();
						for (int i = 0; i < dataStr.length; i++) {
							sb.append(dataStr[i]);
							if (i != dataStr.length - 1) {
								sb.append(",");
							}
						}
						return sb.toString();
					} else {
						return null;
					}
				}
			}).filter(new Function<String, Boolean>() {

				@Override
				public Boolean call(String arg0) throws Exception {
					// TODO Auto-generated method stub
					return arg0 == null ? false : true;
				}
			});
			// 输出格式选择
			switch (mineLabODConfig.outFlag) {
			case 2:
				rdd3.cache().saveAsTextFile(mineLabODConfig.outpath);
				// 生成minemap展示用的数据格式
				JavaPairRDD<String, Iterable<String>> rdd4 = rdd3.mapToPair(new PairFunction<String, String, String>() {

					/** @Fields serialVersionUID: */

					private static final long serialVersionUID = 1L;

					@Override
					public Tuple2<String, String> call(String line) throws Exception {
						// TODO Auto-generated method stub
						String[] arr = line.split(",");
						StringBuffer keyBuf = new StringBuffer();
						StringBuffer valBuf = new StringBuffer();
						keyBuf.append(arr[5]);
						keyBuf.append(",");
						keyBuf.append(arr[11]);
						for (int i = 16; i < 20; i++) {
							valBuf.append(arr[i]);
							if (i != 19) {
								valBuf.append(",");
							}
						}

						return new Tuple2<String, String>(keyBuf.toString(), valBuf.toString());
					}
				}).groupByKey();
				// rdd4.cache().saveAsTextFile(mineLabODConfig.outpath +
				// "_rdd4");
				JavaPairRDD<Integer, Iterable<String>> rdd5 = rdd4.mapToPair(new PairFunction<Tuple2<String, Iterable<String>>, Integer, String>() {

					@Override
					public Tuple2<Integer, String> call(Tuple2<String, Iterable<String>> arg0) throws Exception {
						// TODO Auto-generated method stub
						// 容器一，用来放出发时间戳的od数
						TreeMap<Integer, Integer> outNumMAP = new TreeMap<Integer, Integer>();
						// 容器二，用来放到达时间戳的od数
						TreeMap<Integer, Integer> inNumMAP = new TreeMap<Integer, Integer>();

						int timegranularity = (int) mineLabODConfig.timegranularity;
						int totalTimeSplitNum = 0;
						// System.out.print("timegranularityFlag:"+timegranularityFlag);
						switch (mineLabODConfig.timegranularityFlag) {
						case 0:
							totalTimeSplitNum = 1440 / timegranularity;
							System.out.print("totalTimeSplitNum:" + totalTimeSplitNum);
						}

						// 初始化容器
						for (int j = 0; j < totalTimeSplitNum; j++) {
							outNumMAP.put(j, -1);
							inNumMAP.put(j, -1);
						}

						String key = arg0._1;
						Iterable<String> vals = arg0._2;
						String[] keyArr = key.split(",");
						int o_bestGridId = Integer.parseInt(keyArr[0]);
						int d_bestGridId = Integer.parseInt(keyArr[1]);

						// 获取o点区域中心点、d点区域中心点----此处以后需要修改，目前暂时用二次网格最小粒度
						Envelope oev = MapUtil.getGridBound(o_bestGridId);
						Envelope dev = MapUtil.getGridBound(d_bestGridId);
						double olon = oev.centre().x;
						double olat = oev.centre().y;
						double dlon = dev.centre().x;
						double dlat = dev.centre().y;
						// 获取o点区域形状，d点区域形状
						Polygon oGeo = MapUtil.getMeshPolygon(o_bestGridId);
						Polygon dGeo = MapUtil.getMeshPolygon(d_bestGridId);

						// 进出标志，时间粒度，和od数
						for (String val : vals) {
							String[] valArr = val.split(",");
							if (valArr[0].equals("0")) {
								outNumMAP.put(Integer.parseInt(valArr[2]), Integer.parseInt(valArr[3]));
							} else {
								inNumMAP.put(Integer.parseInt(valArr[2]), Integer.parseInt(valArr[3]));

							}
						}
						if (olon == dlon && olat == dlat) {
							return null;
						}
						StringBuffer returnSb = new StringBuffer();
						returnSb.append(olon + "," + olat + "\t");
						returnSb.append(dlon + "," + dlat + "\t");

						returnSb.append(oGeo.toString() + "\t");
						returnSb.append(dGeo.toString() + "\t");

						StringBuffer inSB = new StringBuffer();
						for (Map.Entry<Integer, Integer> in : inNumMAP.entrySet()) {
							inSB.append(in.getKey() + ":" + in.getValue());
							inSB.append(",");
						}
						String inS = inSB.toString();
						StringBuffer outSB = new StringBuffer();
						for (Map.Entry<Integer, Integer> out : outNumMAP.entrySet()) {
							outSB.append(out.getKey() + ":" + out.getValue());
							outSB.append(",");
						}
						String outS = outSB.toString();

						returnSb.append(inS + "\t");
						returnSb.append(outS + "\t");
						return new Tuple2<Integer, String>(1, returnSb.toString());
					}
				}).filter(new Function<Tuple2<Integer, String>, Boolean>() {

					@Override
					public Boolean call(Tuple2<Integer, String> arg0) throws Exception {
						// TODO Auto-generated method stub
						return arg0 == null ? false : true;
					}
				}).groupByKey(1);
				// rdd5.cache().saveAsTextFile(mineLabODConfig.outpath +
				// "_rdd5");
				JavaRDD<String> rdd6 = rdd5.flatMap(new FlatMapFunction<Tuple2<Integer, Iterable<String>>, String>() {

					@Override
					public Iterable<String> call(Tuple2<Integer, Iterable<String>> arg0) throws Exception {
						// TODO Auto-generated method stub
						ArrayList<String> dataList = new ArrayList<String>();
						int i = 0;
						String head = "ID\tSPoint\tEPoint\tSArea\tEArea\tSTimeWeight\tETimeWeight";
						dataList.add(head);
						Iterable<String> vals = arg0._2;
						for (String val : vals) {
							String nowLine = i + "\t" + val;
							// boolean isGooddata = judgeIsSameODArea(val);
							// if (isGooddata) {
							dataList.add(nowLine);
							i++;
							// }
						}
						return dataList;
					}

					/**
					 * 过滤掉od属于同一点的数据
					 * 
					 * @Description
					 * @author Thinkpad
					 * @param val
					 * @return
					 */
					private boolean judgeIsSameODArea(String line) {
						// TODO Auto-generated method stub
						String[] arr = line.split("\\t");
						System.out.println(line);
						String olonlat = arr[1];
						String dlonlat = arr[2];
						System.out.println(olonlat);
						System.out.println(dlonlat);
						if (!olonlat.equals(dlonlat)) {
							return true;
						}
						return false;
					}
				});
				rdd6.saveAsTextFile(mineLabODConfig.outpath + "_forMineMap");
				break;
			default:
				// 扩展
				break;
			}

			break;

		case "hotSpot":
			// 0代表最小粒度为二次网格
			// 1自定义粒度，目前不规则形状为地图边界图
			JavaPairRDD<String, OD_HotSpot_OutDataBean> hotrdd1 = mineLabODNumBean.flatMapToPair(new ODAnalysisTrajectoryODHotspot(mapBroad, mineLabODConfig.spatialFlag, mineLabODConfig.meshCode, mineLabODConfig.countyCode, mineLabODConfig.cityCode, mineLabODConfig.province, mineLabODConfig.timegranularityFlag, mineLabODConfig.timegranularity, meshCountyMapBroad, mineLabODConfig.spatialgranularityFlag));
			JavaPairRDD<String, OD_HotSpot_OutDataBean> hotrdd2 = hotrdd1.filter(new Function<Tuple2<String, OD_HotSpot_OutDataBean>, Boolean>() {

				@Override
				public Boolean call(Tuple2<String, OD_HotSpot_OutDataBean> arg0) throws Exception {
					// TODO Auto-generated method stub
					return arg0._2 == null ? false : true;
				}
			}).reduceByKey(new Function2<OD_HotSpot_OutDataBean, OD_HotSpot_OutDataBean, OD_HotSpot_OutDataBean>() {

				@Override
				public OD_HotSpot_OutDataBean call(OD_HotSpot_OutDataBean arg0, OD_HotSpot_OutDataBean arg1) throws Exception {
					// TODO Auto-generated method stub
					int onum = arg0.oNum + arg1.oNum;
					int dnum = arg0.dNum + arg1.dNum;
					OD_HotSpot_OutDataBean data = new OD_HotSpot_OutDataBean();
					data.date = arg0.date;
					data.weekday = arg0.weekday;
					data.timeGranularity = arg0.timeGranularity;
					data.timeIndex = arg0.timeIndex;
					data.minGridId = arg0.minGridId;
					data.bestGridId = arg0.bestGridId;
					data.county = arg0.county;
					data.city = arg0.city;
					data.province = arg0.province;
					data.oNum = onum;
					data.dNum = dnum;
					return data;
				}
			});
			JavaRDD<String> hotrdd3 = hotrdd2.map(new Function<Tuple2<String, OD_HotSpot_OutDataBean>, String>() {

				/** @Fields serialVersionUID: */

				private static final long serialVersionUID = 1L;

				@Override
				public String call(Tuple2<String, OD_HotSpot_OutDataBean> arg0) throws Exception {
					// TODO Auto-generated method stub
					arg0.hashCode();
					return arg0._2.toString();
				}
			}).filter(new Function<String, Boolean>() {

				@Override
				public Boolean call(String arg0) throws Exception {
					// TODO Auto-generated method stub
					return arg0 == null ? false : true;
				}
			});
			// 输出格式选择
			switch (mineLabODConfig.outFlag) {
			case 2:
				hotrdd3.saveAsTextFile(mineLabODConfig.outpath);
				break;
			default:
				// 扩展
				break;
			}

			break;
		default:
			break;
		}
	}

	public static void main(String[] args) throws ParseException {

		String userId = "cennavi";
		String modelId = "odModel";
		String modelminId = "hotSpot";
		String dataSourse = "trajectory|trajectory";
		String dataClass = "STL|BAIDU";
		String startDateTime = "20110614000000";
		String endDateTime = "20180614232201";
		int spatialFlag = 1;
		int meshCode = -1;
		int countyCode = -1;
		int cityCode = 110000;
		int provinceCode = -1;
		int timegranularityFlag = 0;
		float timegranularity = 30f;
		int spatialgranularityFlag = 0;
		String geoshapePath = "xiamen_Shape/xiamenjiaotongxiaoqu.SHP";
		int outflag = 2;
		// String outPath =
		// "file:///d:/data/MineLab_lizhizhong_odModel_odNum_trajectory_STL_20160614000000_20160614235959_1487149530558";
		String taskId = getTaskId(userId, modelId, modelminId, dataSourse, dataClass, startDateTime, endDateTime);
		String outPath = "";
		switch (outflag) {
		case 1:
			// 输出到kafka，outFlagAndPath为topic
			outPath = "kafka://" + taskId;
			break;
		case 2:
			// 输出到hdfs
			outPath = "hdfs://nujhadoop/test/minlab/od/" + taskId.replace("|", "_");
			// outPath = "file:///d:/111/" + taskId.replace("|", "_");
			break;
		case 3:
			// 到mysql
			outPath = "mysql://" + taskId;
			break;
		default:

			break;
		}

		MineLabServiceFactory factory = new MineLabServiceFactory();
		IMineLabModel model = factory.getService(userId, modelId, modelminId, dataSourse, dataClass, startDateTime, endDateTime, spatialFlag, meshCode, countyCode, cityCode, provinceCode, timegranularityFlag, timegranularity, spatialgranularityFlag, geoshapePath, outflag, taskId, outPath);
		model.doJob();
	}

	/**
	 * 生成任务id
	 * 
	 * @param modelminId
	 * @param modelId
	 * @param userId
	 * @param outPath
	 * @param endDateTime
	 * @param startDateTime
	 * @param dataClass
	 * @param dataSourse
	 * @return
	 */
	private static synchronized String getTaskId(String userId, String modelId, String modelminId, String dataSourse, String dataClass, String startDateTime, String endDateTime) {
		Date date = new Date();
		String returnStr = userId + "_" + modelId + "_" + modelminId + "_" + dataSourse + "_" + dataClass + "_" + startDateTime + "_" + endDateTime + "_" + date.getTime();
		return returnStr;
	}
}
